package com.diao.datahub;

import com.aliyun.datahub.client.DatahubClient;
import com.aliyun.datahub.client.DatahubClientBuilder;
import com.aliyun.datahub.client.auth.AliyunAccount;
import com.aliyun.datahub.client.common.DatahubConfig;
import com.aliyun.datahub.client.exception.*;
import com.aliyun.datahub.client.http.HttpConfig;
import com.aliyun.datahub.client.model.*;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * @author: chenzhidiao
 * @date: 2021/6/10 10:26
 * @description:
 * @version: 1.0
 */
public class DatahubUtils {

    public static void main(String[] args) {
        DatahubUtils datahub = new DatahubUtils();
        DatahubClient client = datahub.getClient();

//        tupleExample(client,Common.PROJECT,Common.TOPIC);

        getData(client,Common.PROJECT,Common.TOPIC,Common.SUBSCRIPT_ID);
    }
    public static String getTime(){
        //获取系统时间
        Calendar cd = Calendar.getInstance();
        SimpleDateFormat sdf = new SimpleDateFormat("EEE, dd MMM yyyy HH:mm:ss 'GMT'", Locale.US);
        sdf.setTimeZone(TimeZone.getTimeZone("GMT")); // 设置时区为GMT
        return sdf.format(cd.getTime());

    }


    public DatahubClient getClient() {

        String endpoint = Common.ENDPOINT;
        //账号对应的accessId
        String accessId = Common.ACCESS_ID;
        //账号对应的accessKey
        String accessKey = Common.ACCESS_KEY;
        // 创建DataHubClient实例
        DatahubClient datahubClient = DatahubClientBuilder.newBuilder()
                .setDatahubConfig(
                        new DatahubConfig(endpoint,
                                // 是否开启二进制传输，服务端2.12版本开始支持
                                new AliyunAccount(accessId, accessKey), true))
                //专有云使用出错尝试将参数设置为           false
                // HttpConfig可不设置，不设置时采用默认值
                .setHttpConfig(new HttpConfig()
                        .setCompressType(HttpConfig.CompressType.LZ4) // 读写数据推荐打开网络传输 LZ4压缩
                        .setConnTimeout(10000))
                .build();
        return datahubClient;
    }


    /**
     *
     * @param datahubClient 客户端对象
     * @param project 所属项目
     * @param topicName 主题名称
     * @param shardCount 分片数量
     * @param lifeCycle 生命周期
     * @param comment 描述
     */
    public void createTopic(DatahubClient datahubClient,String project,String topicName,int shardCount,int lifeCycle,String comment){
        RecordSchema schema = new RecordSchema();
        schema.addField(new Field("id", FieldType.STRING));

        try {
            CreateTopicResult topic = datahubClient.createTopic(project, topicName, shardCount, lifeCycle, RecordType.TUPLE, schema, comment);
        }catch (Exception e){
            e.printStackTrace();
        }

    }

    // 写入Tuple型数据
    public static void tupleExample(DatahubClient datahubclient, String project, String topic) {

        String shardId = "0";
        // 获取schema
        RecordSchema recordSchema = datahubclient.getTopic(project, topic).getRecordSchema();
        //RecordSchema recordSchema = datahubclient.getTopic(Constant.projectName, Constant.topicName).getRecordSchema();
//        GetProjectResult project = datahubclient.getProject(Constant.projectName);
//        GetTopicResult topic = datahubclient.getTopic(Constant.projectName, Constant.topicName);
        // 生成十条数据
        List<RecordEntry> recordEntries = new ArrayList<RecordEntry>();

        for(int i =20;i<30;i++){
            RecordEntry recordEntry = new RecordEntry();
            // 对每条数据设置额外属性，例如ip 机器名等。可以不设置额外属性，不影响数据写入
            //recordEntry.addAttribute("key1", "value1");
            TupleRecordData data = new TupleRecordData(recordSchema);
            //data.setField("recordtime", time);
            data.setField("userid", "name-"+i);
            //data.setField("message", message);
            data.setField("knowledgeid", "knowledge-"+i);
            data.setField("deptid", "dept-"+i);
            data.setField("createtime", "time:"+System.currentTimeMillis());
            recordEntry.setRecordData(data);
            recordEntry.setShardId(shardId);
            recordEntries.add(recordEntry);
        }

        try {
            int i = datahubclient.putRecords(project, topic, recordEntries).getFailedRecordCount();
            System.out.println(i);
            if (i > 0) {
                retry(datahubclient, recordEntries, 1, project, topic);
            }
        } catch(InvalidParameterException e){
            System.out.println(" invalid parameter, please check your parameter ");
            //logger.error(" invalid parameter, please check your parameter ");
        } catch(AuthorizationFailureException e){
            System.out.println(" AK error, please check your accessId and accessKey ");
            //logger.error(" AK error, please check your accessId and accessKey ");
        } catch(ResourceNotFoundException e){
            System.out.println(" project or topic not found ");
            //logger.error(" project or topic not found ");
        } catch(NoPermissionException e){
            System.out.println(" NoPermission ");
            //logger.error(" NoPermission ");
        } catch(SeekOutOfRangeException e){
            System.out.println(" data is expired or the give timestamp large than current timestamp ");
            //logger.error(" data is expired or the give timestamp large than current timestam ");
        } catch (DatahubClientException e) {
            System.out.println(e.getMessage());
        }
    }

    //重试机制
    public static void retry(DatahubClient client, List<com.aliyun.datahub.client.model.RecordEntry> records, int retryTimes, String project, String topic) {
        boolean suc = false;
        while (retryTimes != 0) {
            retryTimes = retryTimes - 1;
            int failedNum = client.putRecords(project, topic, records).getFailedRecordCount();
            if (failedNum > 0) {
                continue;
            }
            suc = true;
            break;
        }
        if (!suc) {
            System.out.println("retryFailure");
        }
    }


    public static String getURLContent(String strURL) throws Exception {

        URL url = new URL(strURL);
        HttpURLConnection httpConn = (HttpURLConnection) url.openConnection();
        httpConn.setRequestMethod("GET");
        httpConn.connect();

        BufferedReader reader = new BufferedReader(new InputStreamReader(httpConn.getInputStream()));
        String line;
        StringBuffer buffer = new StringBuffer();
        while ((line = reader.readLine()) != null) {
            buffer.append(line);
        }
        reader.close();
        httpConn.disconnect();


        return buffer.toString();
    }


    /**
     *
     * @param datahubClient
     * @param projectName
     * @param topicName
     * @param subId 订阅ID
     */
    public static void getData(DatahubClient datahubClient,String projectName,String topicName,String subId) {
        String shardId = "0";
        List<String> shardIds = Arrays.asList("0");
        OpenSubscriptionSessionResult openSubscriptionSessionResult = datahubClient.openSubscriptionSession(projectName, topicName, subId, shardIds);
        SubscriptionOffset subscriptionOffset = openSubscriptionSessionResult.getOffsets().get(shardId);
        // 1、获取当前点位的cursor，如果当前点位已过期则获取生命周期内第一条record的cursor，未消费同样获取生命周期内第一条record的cursor
        String cursor = null;
        //sequence < 0说明未消费
        if (subscriptionOffset.getSequence() < 0) {
            // 获取生命周期内第一条record的cursor
            cursor = datahubClient.getCursor(projectName, topicName, shardId, CursorType.OLDEST).getCursor();
        } else {
            // 获取下一条记录的Cursor
            long nextSequence = subscriptionOffset.getSequence() + 1;
            try {
                //按照SEQUENCE getCursor可能报SeekOutOfRange错误，表示当前cursor的数据已过期
                cursor = datahubClient.getCursor(projectName, topicName, shardId, CursorType.SEQUENCE, nextSequence).getCursor();
            } catch (SeekOutOfRangeException e) {
                // 获取生命周期内第一条record的cursor
                cursor = datahubClient.getCursor(projectName, topicName, shardId, CursorType.OLDEST).getCursor();
            }
        }
        // 2、读取并保存点位，这里以读取Tuple数据为例，并且每1000条记录保存一次点位
        long recordCount = 0L;
        // 每次读取10条record
        int fetchNum = 10;
        //获取topic的schema信息
        RecordSchema schema = datahubClient.getTopic(projectName, topicName).getRecordSchema();
        while (true) {
            try {
                GetRecordsResult getRecordsResult = datahubClient.getRecords(projectName, topicName, shardId, schema, cursor, fetchNum);
                if (getRecordsResult.getRecordCount() <= 0) {
                    // 无数据，sleep后读取
                    Thread.sleep(1000);
                    continue;
                }
                for (RecordEntry recordEntry : getRecordsResult.getRecords()) {
                    //消费数据
                    TupleRecordData data = (TupleRecordData) recordEntry.getRecordData();
//                    System.out.println("userid:" + data.getField("userid") + "\t"
//                            + "knowledgeid:" + data.getField("knowledgeid") + "\t"
//                            + "deptid:" + data.getField("deptid") + "\t"
//                            + "createtime:" + data.getField("createtime")+ "\t");

                    // 处理数据完成后，设置点位
                    ++recordCount;
                    subscriptionOffset.setSequence(recordEntry.getSequence());
                    subscriptionOffset.setTimestamp(recordEntry.getSystemTime());
                    if (recordCount % 1000 == 0) {
                        //提交点位点位
                        Map<String, SubscriptionOffset> offsetMap = new HashMap<>();
                        offsetMap.put(shardId, subscriptionOffset);
                        datahubClient.commitSubscriptionOffset(projectName, topicName, shardId, offsetMap);
                        System.out.println("commit offset successful");
                    }
                }
                cursor = getRecordsResult.getNextCursor();
            } catch (SubscriptionOfflineException | SubscriptionSessionInvalidException e) {
                // 退出. Offline: 订阅下线; SubscriptionSessionInvalid: 表示订阅被其他客户端同时消费
                break;
            } catch (SubscriptionOffsetResetException e) {
                // 表示点位被重置，重新获取SubscriptionOffset信息，这里以Sequence重置为例
                // 如果以Timestamp重置，需要通过CursorType.SYSTEM_TIME获取cursor
                subscriptionOffset = datahubClient.getSubscriptionOffset(projectName, topicName, shardId, shardIds).getOffsets().get(shardId);
                long nextSequence = subscriptionOffset.getSequence() + 1;
                cursor = datahubClient.getCursor(projectName, topicName, shardId, CursorType.SEQUENCE, nextSequence).getCursor();
            } catch (DatahubClientException e) {
                // TODO: 针对不同异常决定是否退出
            } catch (Exception e) {
                break;
            }
        }
    }
}
